package momo.app.status;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;

import momo.app.config.AppConfig;
import momo.app.structure.Message;
import momo.app.structure.Peer;
import momo.app.util.ConnectionTest;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;


public class LatencyMeasurement extends Thread
{
	private Log log = LogFactory.getLog(LatencyMeasurement.class);
	
	private final int TIME_OUT = 30 * 1000;
	private final int LATENCY_MEASURE_DELAY = 5 * 30 * 1000;
	
	private Peer measuredPeer;
	private Boolean lock;
	private boolean hasAcquiredBefore;
	private double measuredLatency;
	private boolean interrupted;
	
	double expSmoothingParam = 0.2;
	
	public LatencyMeasurement(Peer peer)
	{
		this.measuredPeer = peer;
		lock = false;
		hasAcquiredBefore = false;
		interrupted = false;
		this.start();
	}
	
	public void stopMeasuring()
	{
		interrupted = true;
		synchronized (lock)
		{
			lock.notifyAll();
		}
	}
	
	public void run()
	{
		if ( log.isDebugEnabled() )
			log.debug("Latency Measurement on process");
		
		while (!interrupted)
		{
			try
			{
				double currLatency = acquireLatency();
				if ( log.isDebugEnabled() )
					log.debug("currLatency: me: "+AppConfig.getInstance().getId()+" to : "+measuredPeer.getId()+" = "+currLatency);
				if (Double.isNaN(currLatency)) {
					if ( log.isErrorEnabled() )
						log.error("currLatency NaN (no communication) me: "+AppConfig.getInstance().getId()+" to : "+measuredPeer.getId()+" = "+currLatency);
				}
				else {
					synchronized (lock)
					{
						if (!hasAcquiredBefore)
						{
							measuredLatency = currLatency;
							hasAcquiredBefore = true;
						}else
							measuredLatency = (1 - expSmoothingParam) * measuredLatency + expSmoothingParam * currLatency;
						lock.notifyAll();
					}
					if ( log.isDebugEnabled() )
						log.debug("smooth latency: "+measuredLatency);
				}
			}catch(IOException ie)
			{
				if ( log.isErrorEnabled() )
					log.error("IOException", ie);
				synchronized (lock)
				{
					if ( !hasAcquiredBefore )
						hasAcquiredBefore = true;
					measuredLatency = -1;
					lock.notifyAll();
				}
			}
			
			try 
			{
				Thread.sleep(LATENCY_MEASURE_DELAY);
			}catch (InterruptedException ie) {
				interrupted = true;
			}
		}
	}
	
	private double acquireLatency() throws IOException
	{
		InetAddress address = measuredPeer.getAddressToUse();
		if ( address == null )
		{
			ConnectionTest testConnection = new ConnectionTest(measuredPeer);
			address = testConnection.testConnection();
			measuredPeer.setAddressToUse(address);
		}
		
		if ( measuredPeer.getAddressToUse() == null )
			throw new IOException();
		
		Socket sock = new Socket();
		sock.connect(new InetSocketAddress(measuredPeer.getAddressToUse(), measuredPeer.getStatusPort()), TIME_OUT);
		
		BufferedReader in = new BufferedReader(new InputStreamReader(sock.getInputStream()));
		PrintWriter out = new PrintWriter(sock.getOutputStream(), true);
		
		long start = System.currentTimeMillis();
		
		out.println(
				new Message(Message.NO_FORWARD, 
					AppConfig.getInstance().getId(),
					Message.LATENCY_MEASURE,
					"").toString()
				);
		out.flush();
		int replyChar = in.read();
		
		if (replyChar == -1)
			return Double.NaN;
		
		long end = System.currentTimeMillis();
		
		in.close();
		out.close();
		sock.close();
		
		double calLat = calculateMeasuredLatency(start, end); 
		if ( log.isDebugEnabled() )
			log.debug(AppConfig.getInstance().getId()+"->"+measuredPeer.getId()+": " + calLat + " ms");
		return calLat;
	}
	
	public static double calculateMeasuredLatency(long start, long end)
	{
		long diff = end - start;
		double lat = (double)diff;
		return lat;
	}
	
	public double getMeasuredLatency()
	{
		try
		{
			synchronized(lock)
			{
				if ( !hasAcquiredBefore )
					lock.wait();
			}
		}catch (InterruptedException ie)
		{
			log.error("InterruptedException", ie);
		}
		return measuredLatency;
	}
	
	public Peer getMeasuredPeer()
	{
		return measuredPeer;
	}
	
}//end of class LatencyMeasurement
